package storm.bolt;

import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.BasicOutputCollector;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseBasicBolt;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;

import java.util.Map;

public class OrderDetailSplitBolt extends BaseBasicBolt {
    public void prepare(Map conf, TopologyContext context){

    }
    public void execute(Tuple tuple, BasicOutputCollector collector) {
        //将每一个tuple转为字节码
        byte[]bytes=tuple.getBinary(0);
        //byte->String
        String line = new String(bytes);
        //将每个元组以”，“分割
        String[] str = line.split("\",\"");
        String orderdetail_id =str[0];
        String orderid = str[1];
        String sender  = str[2];
        String sendplace = str[3];
        String recipient = str[4];
        String receiveplace = str[5];
        String  methodofpayment = str[6];
        String paytime= str[7];
        String freight = str[8];
        String total = str[9];

        //System.out.println(line);
        //将信息重组成元组的value传至下一个bolt中进行输出和贮存
        collector.emit(new Values(orderdetail_id,orderid,sender,sendplace,recipient
                ,receiveplace,methodofpayment,paytime,freight,total));
    }
    //向下一个bolt传输元组的key的字段名称与emit配合具体看ppt24、25
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        declarer.declare(new Fields("orderdetail_id", "orderid", "sender","sendplace","recipient",
                "receiveplace","methodofpayment", "paytime","freight","total"));
    }
}

